package concurrent;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;


public class BigFileWordCounter {

    private static final int STATE_INIT = -1;
    private static final int STATE_RUNNING = 0;
    private static final int STATE_OVER = 1;
    private static final int STATE_STOP = 2;
    private static final int STATE_RESULTED = 3;

    private int state = STATE_INIT;

    private long total = 0;

    private BlockingQueue<String> queue = new LinkedBlockingQueue<>( );

    private Thread readingThread;
    private Worker[] workers;
    private int workerCount;

    private String logFilePath;
    private String word;
    private String findingTarget;
    private int wordLength;

    private boolean caseSensitive;

    @SuppressWarnings("unused")
    public BigFileWordCounter(String logFilePath, String word) {
        this(logFilePath, word, 4, true);
    }

    @SuppressWarnings("unused")
    public BigFileWordCounter(String logFilePath, String word, boolean caseSensitive) {
        this(logFilePath, word, 4, caseSensitive);
    }

    public BigFileWordCounter(String logFilePath, String word, int workerCount, boolean caseSensitive) {
        this.logFilePath = logFilePath;
        this.word = word;
        this.caseSensitive = caseSensitive;
        this.findingTarget = this.caseSensitive ? word : word.toLowerCase();
        this.wordLength = word.length();
        this.workerCount = workerCount;
    }

    public void startReading() {
        readingThread = new Thread(this::runReading);
        readingThread.start();
    }


    private void runReading() {
        System.out.println();
        try (BufferedReader reader = new BufferedReader(new FileReader(logFilePath))) {
            String line = reader.readLine();
            while (state < STATE_STOP && line != null) {
                try {
                    queue.put(caseSensitive ? line : line.toLowerCase());
                } catch (InterruptedException ignore) {
                    if (state == STATE_STOP) {
                        break;
                    }
                }
                line = reader.readLine();
            }
            if (state < STATE_OVER) {
                state = STATE_OVER;
            }

            if (workers != null) {
                for (Worker worker : workers) {
                    Thread thread = worker.thread;
                    thread.interrupt();
                }
            }
        } catch (IOException e) {
            throw new IllegalArgumentException(e);
        }
    }

    private CountDownLatch countDownLatch;

    class Worker implements Runnable {
        private long count;
        private Thread thread;


        public Worker() {
            thread = new Thread(this);
        }

        @Override
        public void run() {

            String line = takeLine();
            while (state > STATE_INIT && state < STATE_STOP && line != null) {
                count(line);
                line = takeLine();
            }

            countDownLatch.countDown();
        }


        private void count(String line) {
            int index = line.indexOf(findingTarget);
            while (index > -1) {
                count++;
                index = line.indexOf(findingTarget, index + wordLength);
            }
        }

        void start() {
            thread.start();
        }


        private String takeLine() {
            while (state == STATE_RUNNING || (state == STATE_OVER && !queue.isEmpty())) {
                try {
                    String line = queue.take();
                    if (state == STATE_OVER && !thread.isInterrupted()) {
                        thread.interrupt();
                    }
                    return line;
                } catch (InterruptedException e) {

                }
            }
            return null;
        }
    }


    public void startCounting() {
        countDownLatch = new CountDownLatch(workerCount);
        workers = new Worker[workerCount];
        for (int i = 0; i < workerCount; i++) {
            Worker worker = new Worker();
            workers[i] = worker;
            worker.start();
        }
    }


    public synchronized void start() {

        if (state == STATE_RUNNING) {
            throw new IllegalStateException("the Counter is started!");
        }
        state = STATE_RUNNING;
        this.startCounting();
        this.startReading();

    }

    public synchronized void stop() {
        if (state >= STATE_STOP) {
            throw new IllegalStateException("the Counter is stop!");
        }
        readingThread.interrupt();
    }

    public long total() {
        if (state == STATE_RESULTED) {
            return total;
        }
        boolean interrupted = false;
        synchronized (this) {
            if (state == STATE_RESULTED) {
                return total;
            }
            while (true) {
                try {
                    countDownLatch.await();
                    break;
                } catch (InterruptedException ignore) {
                    interrupted = true;
                }
            }
            for (int i = 0; i < workerCount; i++) {
                total += workers[i].count;
            }
            state = STATE_RESULTED;

        }
        if (interrupted) Thread.currentThread().interrupt();
        return total;
    }


    public long singleThreadCount() {
        total = 0;
        try (BufferedReader reader = new BufferedReader(new FileReader(logFilePath))) {
            String line = reader.readLine();
            while (line != null) {
                line = caseSensitive ? line : line.toLowerCase();
                int index = line.indexOf(findingTarget);
                while (index > -1) {
                    total++;
                    index = line.indexOf(findingTarget, index + wordLength);
                }
                line = reader.readLine();
            }

        } catch (IOException e) {
            throw new IllegalArgumentException(e);
        }
        return total;
    }

    public static void main(String[] args) {
        BigFileWordCounter counter = new BigFileWordCounter("D:\\data\\logs\\united-uss\\all.2019-04-25.log", "main",3, false);
        long based = System.currentTimeMillis();
        System.out.println(based);
        counter.start();
        System.out.println(counter.total());
        System.out.println(System.currentTimeMillis() - based);
        based = System.currentTimeMillis();
        System.out.println(counter.singleThreadCount());
        System.out.println(System.currentTimeMillis() - based);
    }
}
